package com.xx;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.util.HashSet;
import java.util.Set;

public class HdfsApp {
    FileSystem fs = null;

    @Before
    public void init() throws IOException {
//        构造一个配置参数对象,设置一个参数.访问hdfs的URL
        Configuration conf = new Configuration();
//        指定使用hdfs文件系统,hdfs://主机名:8020;
        conf.set("fs.defaultFS", "hdfs://master:8020");
//        进行客户端身份设置
        System.setProperty("HADOOP_USER_NAME", "root");
//        进行客户端身份认证
        fs = org.apache.hadoop.fs.FileSystem.get(conf);
    }

    @Test
//    上传文件
    public void testAddFileToHdfsFromWindows() throws IOException {
//        Windows 下的资源路径.
        Path src = new Path("D:\\12103030117\\hdfs\\src\\test\\java\\com\\xx\\resourse\\testAddFileToHdfsFromWindows" +
                ".txt");
//        hdfs 下的资源路径.
        Path dfs = new Path("/user/root");
//        将Windows 下的资源上传到 hdfs指定的路径下
        fs.copyFromLocalFile(src, dfs);
        fs.close();
    }

    @Test
    public void read() throws IOException {
        String file = "src/test/java/com/xx/resourse/loading.txt";
        File txt = new File(file);
        InputStreamReader read = new InputStreamReader(Files.newInputStream(txt.toPath()));
        BufferedReader br = new BufferedReader(read);
//                将hdfs文件写入指定路径下,再读取该文件.
//        fs.copyToLocalFile(new Path(file), new Path(fileName));
        String line;
        while ((line = br.readLine()) != null) System.out.print(line + "\n");
        read.close();
    }

    @Test
    public void testDownloadFileToWindowsFromHdfs() throws IOException {
        String src = "src/test/java/com/xx/resourse/loading1.txt";
        String dfs = "/user/root/testAddFileToHdfsFromWindows";
        fs.copyToLocalFile(new Path(dfs)
                , new Path(src));
        fs.close();
    }

    @Test
//    创建,删除,重命名文件.
    public void testMkdirAndDeleteAndRename() throws IOException {
//        新建文件夹.多级创建
        fs.mkdirs(new Path("/user/root/testDir/a"));
        fs.mkdirs(new Path("/user/root/testDir/b"));
//        重命名文件夹或者文件(移动也是这样使用).
        fs.rename(new Path("/user/root/testDir"), new Path("/user/root/testDirRename"));
//        删除文件夹,true为非空也删除.
        boolean delete = fs.delete(new Path("/user/root/testDirRename/a"), true);
        System.out.println("删除情况:" + delete);
    }

    //    移动目录和文件.
    @Test
    public void testRemoveDirAndFile() throws IOException {
        fs.rename(new Path("/user/test"), new Path("/user/movetest"));
    }

    @Test
    //    查看文件和目录信息,.
    public void testFileAndDirectory() throws IOException {
        Set<FileStatus> set = new HashSet<>();
        RemoteIterator<LocatedFileStatus> fileStatuses = fs.listLocatedStatus(new Path("/"));
        while (fileStatuses.hasNext()) {
//            FileStatus next = ;
//            System.out.println(next.getPath() + "isDirectory:" + next.isDirectory());
            add(fileStatuses.next(), set);
        }
        set.forEach(e -> System.out.println("文件的地址:" + e.getPath() + "            is directory?:" + e.isDirectory()));
    }

    public void add(FileStatus status, Set<FileStatus> set) throws IOException {
        set.add(status);
        if (status.isDirectory()) {
            RemoteIterator<LocatedFileStatus> fileStatuses = fs.listLocatedStatus(status.getPath());
            while (fileStatuses.hasNext()) {
                add(fileStatuses.next(), set);
            }
        }
    }

    @Test
    public void testString() {
        String s = "hdfs://master:8020/jsdsg/asdfw/sdg";
        String paten = ":8020/";
        String p = s.substring(s.indexOf(paten) + paten.length());
        System.out.println(p);
        int i = p.lastIndexOf('/') + 1;
        String file = i > 0 ? p.substring(i) : p;
        System.out.println(file);

        int last = p.lastIndexOf('/');
        String pp = last > 0 ? p.substring(0, last) : "";
        System.out.println(pp);
//        String pp = p.substring(0, p.lastIndexOf('/'));
//        System.out.println(pp);
    }

    @Test
//    查看目录信息.只显示文件.
    public void testListFilesP() throws IOException {
//        创建一个包含 hdfs目录 的迭代器.
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println(fileStatus.getPath() + "isDirectory:" + fileStatus.isDirectory());
////            打印当前的文件名称.
//            System.out.println("文件名称:" + fileStatus.getPath().getName());
////            打印当前的文件块的大小(都是128m).
//            System.out.println(fileStatus.getBlockSize());
////            打印当前文件的权限.
//            System.out.println(fileStatus.getPermission());
////            获取该文件块的信息(长度,数据快,datanode).
////
//            System.out.println(fileStatus.getLen());
//            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
//            for (BlockLocation bs : blockLocations) {
//                System.out.println(Arrays.toString(bs.getTopologyPaths()));
//                System.out.println("block-length:" + bs.getLength() + "--" + "block-offset:" + bs.getOffset());
//                String[] hosts = bs.getHosts();
//                for (String host : hosts) {
//                    System.out.println(host);
//                }
//                System.out.println("------------------------------");
//            }
        }
    }

    @Test
//    查看文件以及文件夹的信息.
    public void testListAll() throws IOException {
        FileStatus[] listStatus = fs.listStatus(new Path("/"));
        String flag = "d--               ";
        for (FileStatus status : listStatus) {
            if (status.isFile()) flag = "f--              ";
            System.out.println(flag + status.getPath().getName());
        }
    }
}
